networking.gossipsub_subscribe("consensus-0-31", codec).await?;

let (channel1, channel2) = message_channel::channel::<HotstuffMessage>::(100);
networking.gossipsub_subscribe("consensus-0-31", channel1).await?;

pub fn channel<T>(capacity: usize) -> (MessageChannel<T>, MessageChannel<T>) {
    let (ltr_tx, ltr_rx) = mpsc::channel(capacity);
    let (rtl_tx, rtl_rx) = mpsc::channel(capacity);
    (
        MessageChannel {
            sender: ltr_tx,
            receiver: rtl_rx,
        },
        MessageChannel {
            sender: rtl_tx,
            receiver: ltr_rx,
        },
    )
}

#[derive(Debug)]
pub struct MessageChannel<T> {
    sender: mpsc::Sender<T>,
    receiver: mpsc::Receiver<T>,
}

impl<T> MessageChannel<T> {
    pub async fn recv(&mut self) -> Option<T> {
        self.receiver.recv().await
    }

    pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
        self.receiver.try_recv()
    }

    pub async fn send(&self, value: T) -> Result<(), SendError<T>> {
        self.sender.send(value).await
    }

    pub fn try_send(&self, value: T) -> Result<(), TrySendError<T>> {
        self.sender.try_send(value)
    }

  pub fn split(self) -> (mpsc::Receiver<T>, mpsc::Sender<T>) {
        (self.receiver, self.sender)
   } 
}

// Client code
// Set some options for the subscription 
let opts = GossipSubcriptionOpts { max_message_length: 1024*1024 };
// Subscribe - that sets up a channel for sending/receiving
let (sender, receiver) = networking.gossipsub_subscribe("consensus-0-31", opts).await?;
// Receiver goes to inbound messaging
let msg = receiver.next_message().await?; 
// Sender goes to outbound messaging
sender.send(msg).await?;


// In networking

pub async fn gossipsub_subscribe<T: prost::Message>(&self, topic: Topic, opts: GossipSubscriptionOpts) -> Result<...> {
  let (reply, reply_rx) = oneshot::channel();
  // Unbounded because of possible deadlock (consensus is waiting to send to networking, networking is waiting for consensus to read off the channel)
  let (sender, receiver) = mpsc::unbounded_channel();
  self.request_tx.send(NetworkRequest::GossipsubSubscribe {
      topic,
     receiver,
    opts,
   reply,  // Reply<mpsc::Receiver<gossipsub::Message>>
  }).await?;

  let receiver = reply_rx.await?;
  Ok((
    GossipSender::<T>::new(sender),
    GossipReceiver::<T>::new(receiver),
  ))
}

// The subscription types

#[derive(Debug, Clone)]
struct GossipSender<T> {
  sender: mpsc::UnboundedSender<Vec<u8>>,
  codec: ProstCodec<T>,
}

impl<T: prost::Message> GossipSender<T> {
   pub async fn send(&self, msg: T) -> Result<(), MessageError> {
        let len = msg.encoded_len();

       let mut buf = Vec::with_capacity(len);
        self.codec
              .encode_to(&mut buf, message)
              .await?;
        self.sender.send(buf)?;
       Ok(())
  }
}

#[derive(Debug)]
struct GossipReceiver<T> {
  receiver: mpsc::Receiver<(PeerId, gossipsub::Message)>,
  codec: ProstCodec<T>,
}

impl<T: prost::Message> GossipReceiver<T> {
   pub async fn next_message(&self) -> Result<Option<(PeerId, usize, T)>, MessageError> {
          let Some((source, raw_msg)) =  self.receiver.recv().await? else {
            return Ok(None);
         };

        let (_len, msg) = self.codec.decode_from(&mut raw_msg.data.as_slice()).await?;
       
       Ok((source, msg))
   }
}